package core.rdd.example;

import org.apache.poi.ss.usermodel.*;
import org.apache.poi.util.IOUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function;

import java.io.FileInputStream;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;

public class Spark01_HotTop10 {
    public static void main(String[] args) {
        IOUtils.setByteArrayMaxOverride(2000000000);
        /**
         *flatMap 是将多集合转换成一个
         */
        // 配置SparkConf指向你的Spark master URL
        SparkConf conf = new SparkConf()
                .setAppName("Spark01_HotTop10") // 应用名称
                .setMaster("local[*]"); // 替换成你的master地址
        JavaSparkContext sc = new JavaSparkContext(conf);
        // 创建JavaSparkContext，它是与集群交互的主要入口点
        try {
            List<String[]> strings = readExcelFile("data/ods_01_2024_emr_inf_report.xlsx");
            strings.remove(0);



            Path dirPath = Paths.get("patient");
            // 检查文件夹是否存在
            if (Files.exists(dirPath)) {
                // 删除文件夹及其内容
                Files.walk(dirPath)
                        .sorted((path1, path2) -> -path1.compareTo(path2)) // 从子目录到父目录排序
                        .forEach(path -> path.toFile().delete());
            }
            JavaRDD<String[]> parallelize = sc.parallelize(strings, 100);


            JavaRDD<String> objectJavaRDD = parallelize.flatMap(new FlatMapFunction<String[], String>() {
                @Override
                public Iterator<String> call(String[] strings) throws Exception {
                    Iterator<String> iterator = Arrays.asList(strings).iterator();
                    return iterator;
                }
            });

            objectJavaRDD.collect().forEach(System.out::println);


            JavaPairRDD<Object, Iterable<String>> objectIterableJavaPairRDD = objectJavaRDD.groupBy(new Function<String, Object>() {
                @Override
                public Object call(String v1) throws Exception {
                    return null;
                }
            });





        } catch (Exception e) {
            throw new RuntimeException(e);
        } finally {
            sc.close();
        }
    }


    // 方法：使用 Apache POI 读取 Excel 文件
    private static List<String[]> readExcelFile(String filePath) throws Exception {
        List<String[]> data = new ArrayList<>();
        FileInputStream fis = new FileInputStream(filePath);
        Workbook workbook = WorkbookFactory.create(fis);
        Sheet sheet = workbook.getSheetAt(0); // 获取第一个工作表

        Iterator<Row> rowIterator = sheet.iterator();
        while (rowIterator.hasNext()) {
            Row row = rowIterator.next();
            Iterator<Cell> cellIterator = row.cellIterator();

            List<String> rowData = new ArrayList<>();
            while (cellIterator.hasNext()) {
                Cell cell = cellIterator.next();
                switch (cell.getCellType()) {
                    case STRING:
                        rowData.add(cell.getStringCellValue());
                        break;
                    case NUMERIC:
                        if (DateUtil.isCellDateFormatted(cell)) {
                            rowData.add(cell.getDateCellValue().toString());
                        } else {
                            rowData.add(String.valueOf(cell.getNumericCellValue()));
                        }
                        break;
                    case BOOLEAN:
                        rowData.add(String.valueOf(cell.getBooleanCellValue()));
                        break;
                    default:
                        rowData.add("");
                }
            }
            data.add(rowData.toArray(new String[0]));
        }

        workbook.close();
        fis.close();
        return data;
    }
}